package com.lin.mock.pay.server.task;

import cn.hutool.core.lang.UUID;
import cn.hutool.http.HttpRequest;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.lin.mock.pay.server.model.dto.OrderTaskDTO;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

import java.util.Objects;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 任务执行器
 *
 * @author: linjiang
 * @date: 2024/12/16
 */
@Slf4j
public class TaskExcutor implements Runnable {
    private final ConcurrentLinkedQueue<OrderTaskDTO> queue = new ConcurrentLinkedQueue();

    private final ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 10, 60, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(1024));


    /**
     * 获取队列
     *
     * @return
     */
    public ConcurrentLinkedQueue<OrderTaskDTO> getQueue() {
        return queue;
    }

    @Override
    public void run() {
        // 一个永久线程执行。
        while (true) {
            if (queue.size() > 0) {
                try {
                    TimeUnit.SECONDS.sleep(1);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }

                executor.execute(() -> {
                    OrderTaskDTO poll = queue.poll();
                    if (Objects.nonNull(poll)) {
                        Boolean callbackBool = callback(poll);
                        if (!callbackBool) {
                            // 重新投递
                            queue.add(poll);
                        }
                    }
                });
            }
        }
    }

    public Boolean callback(OrderTaskDTO task) {
        JSONObject params = new JSONObject();
        params.put("return_code", "0");
        params.put("out_trade_no", task.getOutTradeNo());
        params.put("attach", task.getAttach());
        params.put("transaction_id", UUID.randomUUID(false).toString());
        try {
            log.info("{} 线程执行支付回调通知【{}】", Thread.currentThread().getName(), params.toJSONString());
            this.invoke(task.getNotifyUrl(), JSON.toJSONString(params));
            return Boolean.TRUE;
        } catch (Exception e) {
            log.error("发送回调通知异常【{}】", e.getMessage());
            return Boolean.FALSE;
        }
    }

    /**
     * 异步调用
     *
     * @param notifyUrl
     * @param params
     */
    @SneakyThrows
    private String invoke(String notifyUrl, String params) {
        HttpRequest httpRequest = HttpRequest.post(notifyUrl);
        httpRequest.body(params);
        httpRequest.setConnectionTimeout(30000);
        httpRequest.setReadTimeout(30000);
        String body = httpRequest.execute().body();
        log.info("执行回调通知得响应结果：{}", body);
        return body;
    }
}
